/** @format */
/* eslint-disable @typescript-eslint/camelcase */

import { Subject, BehaviorSubject, Observable, Observer } from 'rxjs';
import { map, filter, concatAll } from 'rxjs/operators';
import * as fs from 'fs';
import { Readable } from 'stream';

export const TS_PACKET_SIZE = 188; /* 204, 188+16字节的CRC */
export const TS_PAT_PID = 0x00;
export const TS_PAT_TABLE_PID = 0x00;
export const TS_PMT_TABLE_PID = 0x02;
export const TS_H264_PID = 100;
export const TS_ACC_PID = 101;
export const TS_SYSTEM_CLOCK_FREQUENCY = 27000000;

export type TsStreamType = 'UNKNOW' | 'H264' | 'AAC';
export enum TsReaderStatus {
    Idle,
    Working,
}
export enum TsEventType {
    none,
    open,
    close,
    data,
    tsPacket,
    pesPacket,
    esPacket,
    aacPacket,
    h264Packet,
}

export interface TsPATProgram {
    program_number: number; //节目号
    reserved: number; // 保留位
    program_map_PID: number; // 节目映射表的PID，节目号大于0时对应的PID，每个节目对应一个
}

export interface TsPAT {
    table_id: number; //固定为0x00 ，标志是该表是PAT表
    section_syntax_indicator: number; //段语法标志位，固定为1
    zero: number; //0
    reserved_1: number; // 保留位
    section_length: number; //表示从下一个字段开始到CRC32(含)之间有用的字节数
    transport_stream_id: number; //该传输流的ID，区别于一个网络中其它多路复用的流
    reserved_2: number; // 保留位
    version_number: number; //范围0-31，表示PAT的版本号
    current_next_indicator: number; //发送的PAT是当前有效还是下一个PAT有效
    section_number: number; //分段的号码。PAT可能分为多段传输，第一段为00，以后每个分段加1，最多可能有256个分段
    programs: TsPATProgram[]; //最后一个分段的号码
    last_section_number: number;
    reserved_3: number; // 保留位
    network_PID: number; //网络信息表（NIT）的PID,节目号为0时对应的PID为network_PID
    CRC_32: number; //CRC32校验码
}

export interface TsAACPacket {
    time: number;
    pts: number;
    dts: number;
    payload: Buffer;
    adts?: TsADTS;
}

export interface TsH264Packet {
    time: number;
    pts: number;
    dts: number;
    payload: Buffer;
    // sps?: Ts_SPS;
    // pps?: Ts_PPS;
}

export interface TsEvent {
    type: TsEventType;
    buffer?: Buffer;
    tsPacket?: TsPacket;
    pesPacket?: TsPESPacket;
    esPacket?: TsESPacket;
    aacPacket?: TsAACPacket;
    h264Packet?: TsH264Packet;
}

export interface TsPMTStream {
    stream_type: number; //指示特定PID的节目元素包的类型。该处PID由elementary PID指定.流类型，标志是Video还是Audio还是其他数据，h.264编码对应0x1b，aac编码对应0x0f，mp3编码对应0x03
    stringType: TsStreamType;
    elementary_PID: number; //该域指示TS包的PID值。这些TS包含有相关的节目元素
    ES_info_length: number; //前两位bit为00。该域指示跟随其后的描述相关节目元素的byte数
    descriptor: number;
}

export interface TsPESPacket {
    type: TsStreamType;
    packet: Buffer;
}

export interface TsPMT {
    table_id: number; //固定为0x02, 表示PMT表
    section_syntax_indicator: number; //固定为0x01
    zero: number; //0x01
    reserved_1: number; //0x03
    section_length: number; //首先两位bit置为00，它指示段的byte数，由段长度域开始，包含CRC。
    program_number: number; // 指出该节目对应于可应用的Program map PID
    reserved_2: number; //0x03
    version_number: number; //指出TS流中Program map section的版本号
    current_next_indicator: number; //当该位置1时，当前传送的Program map section可用； 当该位置0时，指示当前传送的Program map section不可用，下一个TS流的Program map section有效。
    section_number: number; //固定为0x00
    last_section_number: number; //固定为0x00
    reserved_3: number; //0x07
    PCR_PID: number; //指明TS包的PID值，该TS包含有PCR域， 该PCR值对应于由节目号指定的对应节目。 如果对于私有数据流的节目定义与PCR无关，这个域的值将为0x1FFF。
    reserved_4: number; //预留为0x0F
    program_info_length: number; //前两位bit为00。该域指出跟随其后对节目信息的描述的byte数。
    streams: TsPMTStream[]; //每个元素包含8位, 指示特定PID的节目元素包的类型。该处PID由elementary PID指定
    reserved_5: number; //0x07
    reserved_6: number; //0x0F
    CRC_32: number;
}

export interface TsAdaptation {
    adaptation_field_length: number;
    discontinuity_indicator?: number;
    random_access_indicator?: number;
    elementary_stream_priority_indicator?: number;
    pcr_flag?: number;
    opcr_flag?: number;
    splicing_point_flag?: number;
    transport_private_data_flag?: number;
    adaptation_field_extension_flag?: number;
    program_clock_reference_base?: number;
    program_clock_reference_extension?: number;
    original_program_clock_reference_base?: number;
    original_program_clock_reference_extension?: number;
    splice_countdown?: number;
    transport_private_data_length?: number;
    private_data?: Buffer;
    adaptation_field_extension_length?: number;
    ltw_flag?: number;
    piecewise_rate_flag?: number;
    seamless_splice_flag?: number;
    ltw_valid_flag?: number;
    ltw_offset?: number;
    piecewise_rate?: number;
    splice_type?: number;
    dts_next_au?: number;
}

export interface TsPacket {
    transport_error_indicator: number;
    payload_unit_start_indicator: number;
    transport_priority: number;
    pid: number;
    transport_scrambling_control: number;
    adaptation_field_control: number;
    continuity_counter: number;
    packet: Buffer;
    payload: Buffer;
    adaptation_field?: TsAdaptation;
}

export interface TsPES {
    packet_start_code_prefix: number;
    stream_id: number;
    TsPESPacket_length: number;
    pts_dts_flags: number;
    pes_header_data_length: number;
    pes_pts?: number;
    pes_dts?: number;
}

export interface TsESPacket {
    type: TsStreamType;
    packet: Buffer;
    pts: number;
    dts: number;
}

export interface TsADTS {
    sampling_frequency_index: number;
    audioFrequency: number;
}

// export interface Ts_PPS {}

// export interface Ts_SPS {}

// export interface Ts_VPS {}

function parseAdaptationField(packet: Buffer): TsAdaptation {
    const adaptation_field_length = packet[4];

    if (adaptation_field_length <= 0) {
        const adaptation_field: TsAdaptation = {
            adaptation_field_length: adaptation_field_length,
        };
        return adaptation_field;
    }

    const adaptation_field: TsAdaptation = {
        adaptation_field_length: adaptation_field_length,
        discontinuity_indicator: (packet[5] & 0x80) >>> 7,
        random_access_indicator: (packet[5] & 0x40) >>> 6,
        elementary_stream_priority_indicator: (packet[5] & 0x20) >>> 5,
        pcr_flag: (packet[5] & 0x10) >>> 4,
        opcr_flag: (packet[5] & 0x08) >>> 3,
        splicing_point_flag: (packet[5] & 0x04) >>> 2,
        transport_private_data_flag: (packet[5] & 0x02) >>> 1,
        adaptation_field_extension_flag: packet[5] & 0x01,
    };

    let index = 6;

    if (adaptation_field.pcr_flag) {
        adaptation_field.program_clock_reference_base =
            packet.readUInt32BE(index) * 2 + ((packet[index + 4] & 0x80) >>> 7);

        adaptation_field.program_clock_reference_extension =
            (packet[index + 4] & 0x1) * 256 + (packet[index + 5] >>> 0);

        // let time:number = (adaptation_field.program_clock_reference_base*8589934592*300+adaptation_field.program_clock_reference_extension)/27000000.0;
        // console.log("PCR:",time);
        // debugger;
        index += 6;
    }

    if (adaptation_field.opcr_flag) {
        adaptation_field.original_program_clock_reference_base =
            packet.readUInt32BE(index) * 2 + ((packet[index + 4] & 0x80) >>> 7);

        adaptation_field.original_program_clock_reference_extension =
            (packet[index + 4] & 0x1) * 256 + (packet[index + 5] >>> 0);

        // let time:number = (adaptation_field.original_program_clock_reference_base*8589934592*300+adaptation_field.original_program_clock_reference_extension)/27000000.0;
        // console.log("OPCR:",time);
        // debugger;

        index += 6;
    }

    if (adaptation_field.splicing_point_flag) {
        adaptation_field.splice_countdown = packet.readInt8(index);

        index += 1;
    }

    if (adaptation_field.transport_private_data_flag) {
        adaptation_field.transport_private_data_length = packet[index] >>> 0;

        adaptation_field.private_data = packet.slice(
            index + 1,
            index + 1 + adaptation_field.transport_private_data_length,
        );

        index += 1 + adaptation_field.transport_private_data_length;
    }

    if (adaptation_field.adaptation_field_extension_flag) {
        adaptation_field.adaptation_field_extension_length = packet[index] >>> 0;

        adaptation_field.ltw_flag = (packet[index + 1] & 0x80) >>> 7;
        adaptation_field.piecewise_rate_flag = (packet[index + 1] & 0x40) >>> 6;
        adaptation_field.seamless_splice_flag = (packet[index + 1] & 0x20) >>> 5;

        index += 2;

        if (adaptation_field.ltw_flag) {
            adaptation_field.ltw_valid_flag = (packet[index] & 0x80) >>> 7;

            adaptation_field.ltw_offset = (packet[index] & 0x7f) * 256 + (packet[index + 1] >>> 0);

            index += 2;
        }

        if (adaptation_field.piecewise_rate_flag) {
            adaptation_field.piecewise_rate =
                (packet[index] & 0x3f) * 65536 + packet[index + 1] * 256 + (packet[index + 2] >>> 0);

            index += 3;
        }

        if (adaptation_field.seamless_splice_flag) {
            adaptation_field.splice_type = (packet[index] & 0xf0) >>> 4;

            // 536870912 = 2 ^ 29
            // 4194304   = 2 ^ 22
            // 16384     = 2 ^ 14
            // 128       = 2 ^ 7
            adaptation_field.dts_next_au =
                (packet[index] & 0x0e) * 536870912 +
                packet[index + 1] * 4194304 +
                ((packet[index + 2] & 0xfe) >>> 0) * 16384 +
                packet[index + 3] * 128 +
                ((packet[index + 4] & 0xfe) >>> 1);
        }
    }

    return adaptation_field;
}

function parsePacket(chunk: Buffer): TsPacket {
    if (chunk[0] !== 0x47 /* sync byte */) {
        throw new Error('Not a ts packet');
    }

    const packet: TsPacket = {
        transport_error_indicator: (chunk[1] & 0x80) >>> 7,
        payload_unit_start_indicator: (chunk[1] & 0x40) >>> 6,
        transport_priority: (chunk[1] & 0x20) >>> 5,
        pid: ((chunk[1] & 0x1f) << 8) | chunk[2],
        transport_scrambling_control: (chunk[3] & 0xc0) >>> 6,
        adaptation_field_control: (chunk[3] & 0x30) >>> 4,
        continuity_counter: chunk[3] & 0x0f,
        packet: chunk,
        payload: Buffer.alloc(0),
    };

    // console.log(JSON.stringify({...packet,packet:null,payload:null},undefined,"  "));

    if (packet.adaptation_field_control & 0x2) {
        packet.adaptation_field = parseAdaptationField(chunk);
    }

    if (packet.adaptation_field_control & 0x1) {
        let begin = 5;
        if (packet.adaptation_field) {
            begin += packet.adaptation_field.adaptation_field_length;
        } else if (packet.pid === 256 || packet.pid === 257) {
            begin = 4; //不知道为什么，第五个字节为pes数据
        }
        const end = TS_PACKET_SIZE;
        //for (; chunk[end - 1] === 0xff; --end) {} // 操蛋，这个这个时什么鬼逻辑，适应旧版本么？
        packet.payload = chunk.slice(begin, end);
    }

    return packet;
}

function ParseTsPat(payload: Buffer): TsPAT {
    //00 b00d 0001 c1 00 00 0001 f001 2e701905
    const section_length = ((payload[1] & 0x0f) << 8) | payload[2];
    const ret: TsPAT = {
        table_id: payload[0],
        section_syntax_indicator: payload[1] >> 7,
        zero: (payload[1] >> 6) & 0x1,
        reserved_1: (payload[1] >> 4) & 0x3,
        section_length: ((payload[1] & 0x0f) << 8) | payload[2],
        transport_stream_id: (payload[3] << 8) | payload[4],
        reserved_2: payload[5] >> 6,
        version_number: (payload[5] >> 1) & 0x1f,
        current_next_indicator: (payload[5] << 7) >> 7,
        section_number: payload[6],
        programs: [],
        last_section_number: 0,
        reserved_3: 0,
        network_PID: 0,
        CRC_32:
            ((payload[section_length + 3 - 4] & 0xff) << 24) |
            ((payload[section_length + 3 - 3] & 0xff) << 16) |
            ((payload[section_length + 3 - 2] & 0xff) << 8) |
            (payload[section_length + 3 - 1] & 0xff),
    };
    for (let n = 0; n < section_length - 12; n += 4) {
        const program_num = (payload[8 + n] << 8) | payload[9 + n];
        ret.reserved_3 = payload[10 + n] >> 5;
        ret.network_PID = 0x00;
        if (program_num === 0x00) {
            ret.network_PID = ((payload[10 + n] & 0x1f) << 8) | payload[11 + n];
            // debugger;
            // TS_network_Pid = ret.network_PID; //记录该TS流的网络PID
        } else {
            const program: TsPATProgram = {
                program_number: program_num,
                reserved: payload[10 + n] >> 5,
                program_map_PID: ((payload[10 + n] & 0x1f) << 8) | payload[11 + n],
            };
            // debugger;
            ret.programs.push(program);
            //    TS_program.push_back( PAT_program );//向全局PAT节目数组中添加PAT节目信息
        }
    }
    return ret;
}

function ParseTsPmt(payload: Buffer): TsPMT {
    const section_length = ((payload[1] & 0x0f) << 8) | payload[2];
    const ret: TsPMT = {
        table_id: payload[0],
        section_syntax_indicator: payload[1] >> 7,
        zero: (payload[1] >> 6) & 0x01,
        reserved_1: (payload[1] >> 4) & 0x03,
        section_length: ((payload[1] & 0x0f) << 8) | payload[2],
        program_number: (payload[3] << 8) | payload[4],
        reserved_2: payload[5] >> 6,
        version_number: (payload[5] >> 1) & 0x1f,
        current_next_indicator: (payload[5] << 7) >> 7,
        section_number: payload[6],
        last_section_number: payload[7],
        reserved_3: payload[8] >> 5,
        PCR_PID: ((payload[8] << 8) | payload[9]) & 0x1fff,
        reserved_4: payload[10] >> 4, //预留为0x0F
        program_info_length: ((payload[10] & 0x0f) << 8) | payload[11],
        streams: [],
        reserved_5: 0x07,
        reserved_6: 0x0f,
        CRC_32:
            ((payload[section_length + 3 - 4] & 0xff) << 24) |
            ((payload[section_length + 3 - 3] & 0xff) << 16) |
            ((payload[section_length + 3 - 2] & 0xff) << 8) |
            (payload[section_length + 3 - 1] & 0xff),
    };
    //  PCRID = packet->PCR_PID;

    let pos = 12;
    // program info descriptor
    if (ret.program_info_length !== 0) pos += ret.program_info_length;

    // Get stream type and PID
    for (; pos <= ret.section_length + 2 - 4; ) {
        const pmt_stream: TsPMTStream = {
            stream_type: payload[pos], //指示特定PID的节目元素包的类型。该处PID由elementary PID指定
            stringType: 'UNKNOW',
            elementary_PID: ((payload[pos + 1] << 8) | payload[pos + 2]) & 0x1fff, //该域指示TS包的PID值。这些TS包含有相关的节目元素
            ES_info_length: ((payload[pos + 3] & 0x0f) << 8) | payload[pos + 4], //前两位bit为00。该域指示跟随其后的描述相关节目元素的byte数
            descriptor: 0x00,
        };

        switch (pmt_stream.stream_type) {
            case 0xcf: // SAMPLE-AES AAC
                console.log('SAMPLE-AES AAC not supported');
                break;
            case 0x0f: // ISO/IEC 13818-7 ADTS AAC (MPEG-2 lower bit-rate audio)
                // console.log(0x0f,'ADTS AAC');
                pmt_stream.stringType = 'AAC';
                break;
            case 0x15: // Packetized metadata (ID3)
                console.log('ID3 not supported');
                break;
            case 0xdb: // SAMPLE-AES AVC
                console.log('SAMPLE-AES AVC not supported');
                break;
            case 0x1b: // ITU-T Rec. H.264 and ISO/IEC 14496-10 (lower bit-rate video)
                // console.log(0x1b,'AVC H.264');
                pmt_stream.stringType = 'H264';
                break;
            case 0x03: //ISO/IEC 11172-3 (MPEG-1 audio)
                console.log('MPEG-1 audio not supported');
                break;
            case 0x04: //ISO/IEC 13818-3 (MPEG-2 halved sample rate audio)
                console.log('MPEG-2 halved sample rate audio not supported');
                break;
            case 0x24:
                console.log('HEVC H.265 not supported');
                break;
            default:
                console.log('unkown stream_type:' + pmt_stream.stream_type);
                break;
        }

        ret.reserved_5 = payload[pos + 1] >> 5;
        ret.reserved_6 = payload[pos + 3] >> 4;

        if (pmt_stream.ES_info_length !== 0) {
            pmt_stream.descriptor = payload[pos + 5];

            for (let len = 2; len <= pmt_stream.ES_info_length; len++) {
                pmt_stream.descriptor = (pmt_stream.descriptor << 8) | payload[pos + 4 + len];
            }
            pos += pmt_stream.ES_info_length;
        }
        pos += 5;
        if (pmt_stream.stringType !== 'UNKNOW') {
            ret.streams.push(pmt_stream);
        }
        // TS_Stream_type.push_back(pmt_stream);
    }
    return ret;
}

function parsePES(packet: TsPESPacket): TsESPacket {
    const esPacket: TsESPacket = {
        type: packet.type,
        packet: Buffer.alloc(0),
        pts: 0,
        dts: 0,
    };

    const pesPacket: TsPES = {
        packet_start_code_prefix:
            ((packet.packet[0] & 0xff) << 16) | ((packet.packet[1] & 0xff) << 8) | (packet.packet[2] & 0xff),
        stream_id: packet.packet[3] & 0xff,
        TsPESPacket_length: ((packet.packet[4] & 0xff) << 8) | (packet.packet[5] & 0xff),
        pts_dts_flags: (packet.packet[7] & 0xc0) >> 6,
        pes_header_data_length: packet.packet[8],
    };

    // console.log(packet.packet);

    if (pesPacket.packet_start_code_prefix === 1) {
        if (pesPacket.TsPESPacket_length && pesPacket.TsPESPacket_length !== packet.packet.length - 6) {
            // console.log("-----------------------------");
            console.log('error pes length');
            console.log(
                pesPacket.TsPESPacket_length,
                packet.packet.length - 6,
                packet.packet.length - 6 - pesPacket.TsPESPacket_length,
            );
            // console.log("-----------------------------");
            // console.log(pes.slice(0,64).toString('hex'));
            // console.log("-----------------------------");
            debugger;
            return esPacket;
        }

        if (pesPacket.pts_dts_flags & 0x02) {
            /* PES header described here : http://dvd.sourceforge.net/dvdinfo/pes-hdr.html
                as PTS / DTS is 33 bit we cannot use bitwise operator in JS,
                as Bitwise operators treat their operands as a sequence of 32 bits */
            pesPacket.pes_pts =
                (packet.packet[9] & 0x0e) * 536870912 + // 1 << 29
                (packet.packet[10] & 0xff) * 4194304 + // 1 << 22
                (packet.packet[11] & 0xfe) * 16384 + // 1 << 14
                (packet.packet[12] & 0xff) * 128 + // 1 << 7
                (packet.packet[13] & 0xfe) / 2;
            // check if greater than 2^32 -1
            if (pesPacket.pes_pts > 4294967295) {
                // decrement 2^33
                pesPacket.pes_pts -= 8589934592;
            }
            if (pesPacket.pts_dts_flags & 0x01) {
                pesPacket.pes_dts =
                    (packet.packet[14] & 0x0e) * 536870912 + // 1 << 29
                    (packet.packet[15] & 0xff) * 4194304 + // 1 << 22
                    (packet.packet[16] & 0xfe) * 16384 + // 1 << 14
                    (packet.packet[17] & 0xff) * 128 + // 1 << 7
                    (packet.packet[18] & 0xfe) / 2;
                // check if greater than 2^32 -1
                if (pesPacket.pes_dts > 4294967295) {
                    // decrement 2^33
                    pesPacket.pes_dts -= 8589934592;
                }
                if (pesPacket.pes_pts - pesPacket.pes_dts > 60 * 90000) {
                    console.log(
                        `${Math.round(
                            (pesPacket.pes_pts - pesPacket.pes_dts) / 90000,
                        )}s delta between PTS and DTS, align them`,
                    );
                    pesPacket.pes_pts = pesPacket.pes_dts;
                }
            } else {
                pesPacket.pes_dts = pesPacket.pes_pts;
            }
            esPacket.dts = pesPacket.pes_dts;
            esPacket.pts = pesPacket.pes_pts;
        }
        // 9 bytes : 6 bytes for PES header + 3 bytes for PES extension
        const payloadStartOffset = pesPacket.pes_header_data_length + 9;
        esPacket.packet = packet.packet.slice(payloadStartOffset);
    }
    // console.log(pesPacket.pts_dts_flags, pesPacket.TsPESPacket_length, pesPacket.pes_header_data_length,pesPacket.pes_pts,pesPacket.pes_dts);

    // debugger;
    return esPacket;
}

// function parseADTS(adts: Buffer): TsADTS {
//     //TODO 哥累了，懒得解析了，仅仅取出有用字段
//     const ret = { sampling_frequency_index: 8, audioFrequency: 16000 };
//     if (adts.length !== 7) {
//         return ret;
//     }
//     ret.sampling_frequency_index = (adts[2] & 0x3c) >> 2;
//     switch (ret.sampling_frequency_index) {
//         case 0:
//             ret.audioFrequency = 96000;
//             break;
//         case 1:
//             ret.audioFrequency = 88200;
//             break;
//         case 2:
//             ret.audioFrequency = 64000;
//             break;
//         case 3:
//             ret.audioFrequency = 48000;
//             break;
//         case 4:
//             ret.audioFrequency = 44100;
//             break;
//         case 5:
//             ret.audioFrequency = 32000;
//             break;
//         case 6:
//             ret.audioFrequency = 24000;
//             break;
//         case 7:
//             ret.audioFrequency = 22050;
//             break;
//         case 8:
//             ret.audioFrequency = 16000;
//             break;
//         case 9:
//             ret.audioFrequency = 12000;
//             break;
//         case 10:
//             ret.audioFrequency = 11025;
//             break;
//         case 11:
//             ret.audioFrequency = 8000;
//             break;
//         case 12:
//             ret.audioFrequency = 7350;
//             break;
//         case 13:
//         case 14:
//         case 15:
//         default:
//             break;
//     }
//     return ret;
// }

export class TsReader {
    //File
    private fileStream: fs.ReadStream | undefined;
    private status: TsReaderStatus = TsReaderStatus.Idle;
    private buffer: Buffer = Buffer.alloc(0);
    private files: string[] = [];
    public readonly subjectTsPacket: Subject<TsEvent> = new Subject<TsEvent>();
    public readonly subjectStatus: BehaviorSubject<TsReaderStatus> = new BehaviorSubject<TsReaderStatus>(
        TsReaderStatus.Idle,
    );
    //PES
    private patData: TsPAT | undefined = undefined;
    private pmtPID = -1;
    private pmtData: TsPMT | undefined = undefined;
    private pesCache: Buffer = Buffer.alloc(0);
    private currentStreamType: TsStreamType = 'UNKNOW';
    public subjectPesPacket: Observable<TsEvent>;
    //ES
    private audioFrequency = -1;
    private videoFrequency = -1;
    public readonly subjectEsPacket: Observable<TsEvent>;
    //constructor
    constructor() {
        this.subjectPesPacket = this.subjectTsPacket.pipe(
            map(
                (event: TsEvent): Observable<TsEvent> => {
                    // if(event.buffer) console.log(event.buffer.toString('hex'));
                    return Observable.create((observer: Observer<TsEvent>): void => {
                        if (event.type !== TsEventType.data) {
                            observer.next(event);
                            observer.complete();
                            return;
                        } else {
                            if (event.buffer) {
                                try {
                                    observer.next({ type: TsEventType.tsPacket, tsPacket: parsePacket(event.buffer) });
                                } catch (e) {
                                    observer.error(e);
                                }
                            }
                            observer.complete();
                            return;
                        }
                    });
                },
            ),
            concatAll(),
            map(
                (event: TsEvent): Observable<TsEvent> => {
                    // console.log(event.tsPacket);
                    return Observable.create((observer: Observer<TsEvent>): void => {
                        if (event.type === TsEventType.close) {
                            observer.next({
                                type: TsEventType.pesPacket,
                                pesPacket: {
                                    type: this.currentStreamType,
                                    packet: this.pesCache,
                                },
                            });
                            this.clear();
                            observer.complete();
                            return;
                        }
                        if (event.type === TsEventType.tsPacket && event.tsPacket) {
                            // console.log(event.tsPacket.payload.toString('hex'));
                            const tsData = event.tsPacket;
                            if (!this.patData && tsData.pid === TS_PAT_PID && tsData.payload[0] === TS_PAT_TABLE_PID) {
                                // console.log("PAT==>");
                                // console.log("-----------------------------");
                                // console.log(tsData.packet.toString('hex'));
                                // console.log(tsData.payload.toString('hex'));
                                // console.log("-----------------------------");
                                // console.log("PID:" + tsData.pid);
                                // console.log("AFC:", tsData.adaptation_field_control);
                                this.patData = ParseTsPat(tsData.payload);
                                for (let i = 0; i < this.patData.programs.length; i++) {
                                    const element = this.patData.programs[i];
                                    // console.log(element);
                                    this.pmtPID = element.program_map_PID;
                                }
                                // debugger;
                                observer.complete();
                                return;
                            }

                            if (this.patData && tsData.pid === this.pmtPID && tsData.payload[0] === TS_PMT_TABLE_PID) {
                                // console.log("PMT==>");
                                // console.log("-----------------------------");
                                // console.log(tsData.packet.toString('hex'));
                                // console.log(tsData.payload.toString('hex'));
                                // console.log("-----------------------------");
                                // console.log("PID:", tsData.pid);
                                // console.log("AFC:", tsData.adaptation_field_control);
                                this.pmtData = ParseTsPmt(tsData.payload);
                                // debugger;
                                observer.complete();
                                return;
                            }

                            if (this.patData && this.pmtData) {
                                const stremaType = this.getStreamType(tsData.pid);
                                if (stremaType === 'UNKNOW') {
                                    observer.complete();
                                    return;
                                }
                                if (tsData.payload_unit_start_indicator === 1) {
                                    if (this.pesCache.length > 0) {
                                        observer.next({
                                            type: TsEventType.pesPacket,
                                            pesPacket: {
                                                type: this.currentStreamType,
                                                packet: this.pesCache,
                                            },
                                        });
                                    }
                                    this.pesCache = Buffer.alloc(0);
                                    this.currentStreamType = stremaType;
                                }
                                if (this.currentStreamType !== stremaType) {
                                    console.log('error streamType');
                                    observer.complete();
                                    return;
                                }

                                // console.log("DATA==>");
                                this.pesCache = Buffer.concat([this.pesCache, tsData.payload]);
                                // if(tsData.payload_unit_start_indicator || tsData.payload.length !== 184){
                                // console.log("-----------------------------");
                                // console.log(tsData.packet.toString('hex'));
                                // console.log(tsData.payload.toString('hex'));
                                // console.log("-----------------------------");
                                // console.log("start",tsData.payload_unit_start_indicator,"PID:",tsData.pid," AFC:",tsData.adaptation_field_control," pesLength:",this.pesCache.length,tsData.payload.length);
                                // }
                                observer.complete();
                                return;
                            }
                        }
                        observer.complete();
                        return;
                    });
                },
            ),
            concatAll(),
        );
        this.subjectEsPacket = this.subjectPesPacket.pipe(
            // tap(value=>console.log(value.pesPacket)),
            map(
                (event: TsEvent): Observable<TsEvent> => {
                    return Observable.create((observer: Observer<TsEvent>): void => {
                        if (event.type === TsEventType.pesPacket && event.pesPacket) {
                            observer.next({
                                type: TsEventType.esPacket,
                                esPacket: parsePES(event.pesPacket),
                            });
                        }
                        observer.complete();
                        return;
                    });
                },
            ),
            concatAll(),
            map(
                (event: TsEvent): Observable<TsEvent> => {
                    return Observable.create((observer: Observer<TsEvent>): void => {
                        if (event.type === TsEventType.esPacket && event.esPacket) {
                            if (event.esPacket.type === 'AAC') {
                                observer.next({
                                    type: TsEventType.aacPacket,
                                    aacPacket: {
                                        time: (event.esPacket.pts * 1000) / 90000, //传说：pts、dts 是以 timescale 为时间单位的, 1s = 90000 timescale
                                        pts: event.esPacket.pts,
                                        dts: event.esPacket.dts,
                                        payload: event.esPacket.packet, //TODO 直接发带ADTS的包应该解码器可自适应
                                    },
                                });
                            } else if (event.esPacket.type === 'H264') {
                                observer.next({
                                    type: TsEventType.h264Packet,
                                    h264Packet: {
                                        time: (event.esPacket.pts * 1000) / 90000, //传说：pts、dts 是以 timescale 为时间单位的, 1s = 90000 timescale
                                        pts: event.esPacket.pts,
                                        dts: event.esPacket.dts,
                                        payload: event.esPacket.packet.slice(6), //删掉0000000109XX
                                    },
                                });
                            }
                        }
                        observer.complete();
                    });
                },
            ),
            concatAll(),
            filter((event: TsEvent): boolean => {
                if (event.type === TsEventType.h264Packet) return true;
                else return false;
            }),
        );
    }
    //public
    public appendFiles(files: string[]): void {
        this.files = [...this.files, ...files];
        this.start();
    }
    //private
    private clear(): void {
        this.patData = undefined;
        this.pmtPID = -1;
        this.pmtData = undefined;
        this.pesCache = Buffer.alloc(0);
        this.currentStreamType = 'UNKNOW';
    }
    private getStreamType(pid: number): TsStreamType {
        if (this.pmtData) {
            for (let i = 0; i < this.pmtData.streams.length; i++) {
                if (this.pmtData.streams[i].elementary_PID === pid) {
                    return this.pmtData.streams[i].stringType;
                }
            }
        }
        return 'UNKNOW';
    }
    private openNextFileStream(): void {
        if (this.files.length > 0) {
            const file = this.files.shift();
            if (file) {
                console.log(file);
                this.fileStream = fs.createReadStream(file, { highWaterMark: 188 * 100 });
                this.fileStream.on('open', (fd: number) => {
                    console.log('openFile:' + fd);
                    this.subjectTsPacket.next({ type: TsEventType.open });
                });
                this.fileStream.on('close', () => {
                    this.fileStream = undefined;
                    this.subjectTsPacket.next({ type: TsEventType.close });
                    process.nextTick(() => {
                        this.openNextFileStream();
                    });
                });
                const readStream: Readable = this.fileStream;
                readStream.on('data', (chunk: Buffer) => {
                    // console.log(this.buffer.length,chunk.length);
                    this.buffer = Buffer.concat([this.buffer, chunk]);
                    let index = 0;
                    for (; index + TS_PACKET_SIZE <= this.buffer.length; index += TS_PACKET_SIZE) {
                        this.subjectTsPacket.next({
                            type: TsEventType.data,
                            buffer: this.buffer.slice(index, index + TS_PACKET_SIZE),
                        });
                    }
                    // console.log(this.buffer.length,index);
                    this.buffer = this.buffer.slice(index);
                    // console.log(this.buffer.length);
                });
                this.setStatus(TsReaderStatus.Working);
            }
        } else {
            this.setStatus(TsReaderStatus.Idle);
        }
    }
    private setStatus(status: TsReaderStatus): void {
        this.status = status;
        this.subjectStatus.next(status);
    }
    private start(): boolean {
        if (this.status === TsReaderStatus.Idle) {
            if (this.files.length > 0) {
                this.openNextFileStream();
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }
}
